package com.wuwangfu.exactlyonce;

import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.Charset;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-07-02
 */
public class KafkaStringSerializationSchema implements KafkaSerializationSchema<String> {
    private String topic;
    private String charset;

    //构造方法，传入要写入的topic和字符集，默认使用UTF-8
    public KafkaStringSerializationSchema(String topic) {this(topic,"UTF-8");}
    public KafkaStringSerializationSchema(String topic, String charset) {
        this.topic = topic;
        this.charset = charset;
    }

    /**
     * 调用该方法将数据进行序列化
     *
     * @param element element to be serialized
     * @param timestamp timestamp (can be null)
     * @return
     */
    @Override
    public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
        //将数据转成bytes数组
        byte[] bytes = element.getBytes(Charset.forName(charset));
        //返回ProducerRecord
        return new ProducerRecord<>(topic,bytes);
    }
}
